package com.atguigu.app.dwm;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONAware;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.utils.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.PatternTimeoutFunction;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.OutputTag;

import java.time.Duration;
import java.util.List;
import java.util.Map;

public class UserJumpDetailApp {

    public static void main(String[] args) throws Exception {

        //TODO 1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //设置CK & 状态后端
        //env.enableCheckpointing(5000L);
        //env.getCheckpointConfig().setCheckpointTimeout(5000L);
        //env.setStateBackend(new FsStateBackend("hdfs://hadoop102:8020/flink-210108/cdc/ck"));

        //TODO 2.读取Kafka数据创建流并转换为JSONObject
        String sourceTopic = "dwd_page_log";
        String groupId = "userJumpDetailApp_210108";
        String sinkTopic = "dwm_user_jump_detail";
        SingleOutputStreamOperator<JSONObject> jsonObjDS = env.addSource(MyKafkaUtil.getKafkaConsumer(sourceTopic, groupId))
                .map(JSONObject::parseObject);

        //TODO 3.提取事件时间生成WaterMark
        SingleOutputStreamOperator<JSONObject> jsonObjWithWMDS = jsonObjDS.assignTimestampsAndWatermarks(WatermarkStrategy.<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(1)).withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
            @Override
            public long extractTimestamp(JSONObject element, long recordTimestamp) {
                return element.getLong("ts");
            }
        }));

//        jsonObjWithWMDS.print("jsonObjWithWMDS>>>>>>");

        //TODO 4.按照mid进行分组
        KeyedStream<JSONObject, String> keyedStream = jsonObjWithWMDS
                .keyBy(data -> data.getJSONObject("common").getString("mid"));

        //TODO 5.定义模式序列
        Pattern<JSONObject, JSONObject> pattern = Pattern.<JSONObject>begin("start").where(new SimpleCondition<JSONObject>() {
            @Override
            public boolean filter(JSONObject value) throws Exception {
                String lastPage = value.getJSONObject("page").getString("last_page_id");
                return lastPage == null || lastPage.length() <= 0;
            }
        }).next("next").where(new SimpleCondition<JSONObject>() {
            @Override
            public boolean filter(JSONObject value) throws Exception {
                String lastPage = value.getJSONObject("page").getString("last_page_id");
                return lastPage == null || lastPage.length() <= 0;
            }
        }).within(Time.seconds(10));

//        Pattern<JSONObject, JSONObject> pattern1 = Pattern.<JSONObject>begin("start").where(new SimpleCondition<JSONObject>() {
//            @Override
//            public boolean filter(JSONObject value) throws Exception {
//                String lastPage = value.getJSONObject("page").getString("last_page_id");
//                return lastPage == null || lastPage.length() <= 0;
//            }
//        })
//                .times(2)
//                .consecutive()      //循环模式默认为宽松近邻,加上该条件,则改为严格近邻
//                .within(Time.seconds(10));

        //TODO 6.将模式序列作用到流上
        PatternStream<JSONObject> patternStream = CEP.pattern(keyedStream, pattern);

        //TODO 7.提取事件,匹配上的和超时事件都需要提取
        OutputTag<JSONObject> outputTag = new OutputTag<JSONObject>("TimeOut") {
        };
        SingleOutputStreamOperator<JSONObject> selectDS = patternStream.select(outputTag,
                new PatternTimeoutFunction<JSONObject, JSONObject>() {
                    @Override
                    public JSONObject timeout(Map<String, List<JSONObject>> map, long l) throws Exception {
                        //提取超时事件
                        List<JSONObject> startList = map.get("start");
                        return startList.get(0);
                    }
                }, new PatternSelectFunction<JSONObject, JSONObject>() {
                    @Override
                    public JSONObject select(Map<String, List<JSONObject>> map) throws Exception {
                        //提取匹配上的事件
                        List<JSONObject> startList = map.get("start");
                        return startList.get(0);
                    }
                });

        //TODO 8.将数据写入Kafka
        DataStream<JSONObject> timeOutDS = selectDS.getSideOutput(outputTag);
        DataStream<JSONObject> unionDS = selectDS.union(timeOutDS);

        unionDS.print(">>>>>>>>>");
        unionDS.map(JSONAware::toJSONString).addSink(MyKafkaUtil.getKafkaProducer(sinkTopic));

        //TODO 9.启动
        env.execute();

    }
}
